package com.huawei.dms.rocketmq.consumergroup;

import com.huaweicloud.sdk.core.auth.BasicCredentials;
import com.huaweicloud.sdk.core.auth.ICredential;
import com.huaweicloud.sdk.rocketmq.v2.RocketMQClient;
import com.huaweicloud.sdk.rocketmq.v2.model.CreateConsumerGroupOrBatchDeleteConsumerGroupReq;
import com.huaweicloud.sdk.rocketmq.v2.model.CreateConsumerGroupOrBatchDeleteConsumerGroupRequest;
import com.huaweicloud.sdk.rocketmq.v2.model.CreateConsumerGroupOrBatchDeleteConsumerGroupResponse;
import com.huaweicloud.sdk.rocketmq.v2.model.DeleteConsumerGroupRequest;
import com.huaweicloud.sdk.rocketmq.v2.model.DeleteConsumerGroupResponse;
import com.huaweicloud.sdk.rocketmq.v2.model.ShowGroupRequest;
import com.huaweicloud.sdk.rocketmq.v2.model.ShowGroupResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;

public class RocketMQConsumerGroupDemo {
    private static final Logger logger = LoggerFactory.getLogger(RocketMQConsumerGroupDemo.class);

    public static void main(String[] args) {
        String ak = "<YOUR ak>";
        String sk = "<YOUR sk>";
        String endpoint = "<YOUR Endpoint String>";
        String projectId = "<YOUR Project Id>";
        ICredential auth = new BasicCredentials()
                .withAk(ak)
                .withSk(sk)
                .withProjectId(projectId);
        RocketMQClient client = RocketMQClient.newBuilder()
                .withCredential(auth)
                .withEndpoint(endpoint)
                .build();
        // 创建消费组
        createConsumerGroup(client);
        // 查询消费组
        showGroup(client);
        // 删除消费组
        deleteConsumerGroup(client);
    }

    private static void createConsumerGroup(RocketMQClient client) {

        try {
            CreateConsumerGroupOrBatchDeleteConsumerGroupRequest request = new CreateConsumerGroupOrBatchDeleteConsumerGroupRequest();
            CreateConsumerGroupOrBatchDeleteConsumerGroupReq body = new CreateConsumerGroupOrBatchDeleteConsumerGroupReq();
            request.withInstanceId("<YOUR InstanceId>");
            // 消费组 名称
            body.withName("test-consumerGroup");
            // 代理
            List<String> brokers = new ArrayList<>();
            brokers.add("broker-0");
            body.withBrokers(brokers);
            // 是否广播消费
            body.withBroadcast(false);
            // 最大重试次数
            body.withRetryMaxTime(new BigDecimal(16));
            // 是否重头消费
            body.withFromBeginning(false);
            request.withBody(body);
            CreateConsumerGroupOrBatchDeleteConsumerGroupResponse response = client.createConsumerGroupOrBatchDeleteConsumerGroup(request);
            logger.info(String.valueOf(response.toString()));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private static void deleteConsumerGroup(RocketMQClient client) {
        try {
            DeleteConsumerGroupRequest request = new DeleteConsumerGroupRequest();
            request.withInstanceId("<YOUR InstanceId>");
            request.withGroup("test-consumerGroup");
            DeleteConsumerGroupResponse response = client.deleteConsumerGroup(request);
            logger.info(String.valueOf(response.toString()));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private static void showGroup(RocketMQClient client) {
        try {
            ShowGroupRequest request = new ShowGroupRequest();
            request.withInstanceId("<YOUR InstanceId>");
            request.withGroup("test-consumerGroup");
            ShowGroupResponse response = client.showGroup(request);
            logger.info(String.valueOf(response.toString()));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}
